package com.hellorxjava.demo;

import com.hellorxjava.bean.Animal;
import com.hellorxjava.bean.Dog;

import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.List;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableSource;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.core.SingleObserver;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.BiFunction;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.observables.GroupedObservable;

/**
 * Observer.buffer(int count, int skip)  每次发送一个包含count个数据的集合需要跳过skip个数据   注意是从 count的开始算 1 count >= skip的话没有跳的效果
 *
 * flatMap 将我们想要的数据转换后再发出去 对顺序有要求的话用concatMap
 * groupBy 按照key来拆分一些小的observable 然后这些小的Observable分别发送其包含的数据
 * map 将源Observable发送的每个数据都按照给定的函数进行转化并且将转化后的数据发送出来
 * cast 将发送的数据类型强转为另一种数据类型
 * scan 操作符对一个序列的数据同一个函数进行计算，并将这个结果发送出来 也就是要发送的大数据都是上一次发送出来的数据和这次源Observable的数据作为输入，应用到一个函数计算
 * window 将数据发送多个小的Observable对象然后将其内部的数据发送出去
 */
public class RXJavaDemo5 {
    public static void main(String[] args) {
        bufferObserver().subscribe(new Observer<List<Integer>>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("buffer subscribe");
            }

            @Override
            public void onNext(@NonNull List<Integer> integers) {
                System.out.println("buffer onNext: " + integers);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                System.out.println("buffer onError: " + e.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("buffer onComplete: ");
            }
        });

        flatMapObserver().subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("flatMap subscribe");
            }

            @Override
            public void onNext(@NonNull String s) {
                System.out.println("flatMap onNext: " + s);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                System.out.println("flatMap onError: " + e.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("flatMap onComplete: ");
            }
        });


        flatMapIterableObserver().subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("flatMapIterable subscribe");
            }

            @Override
            public void onNext(@NonNull String s) {
                System.out.println("flatMapIterable onNext: " + s);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                System.out.println("flatMapIterable onError: " + e.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("flatMapIterable onComplete: ");
            }
        });

        groupByObserver().subscribe(new Observer<GroupedObservable<Integer, Integer>>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("groupBy subscribe");
            }

            @Override
            public void onNext(@NonNull final GroupedObservable<Integer, Integer> observable) {
//                observable.count().subscribe(new SingleObserver<Long>() {
//                    @Override
//                    public void onSubscribe(@NonNull Disposable d) {
//
//                    }
//
//                    @Override
//                    public void onSuccess(@NonNull Long aLong) {
//                        System.out.println("key: " + observable.getKey() + " contains：" + aLong + "numbers");
//                    }
//
//                    @Override
//                    public void onError(@NonNull Throwable e) {
//
//                    }
//                });

                if(observable.getKey() == 0) {
                    observable.subscribe(new Observer<Integer>() {
                        @Override
                        public void onSubscribe(@NonNull Disposable d) {

                        }

                        @Override
                        public void onNext(@NonNull Integer integer) {
                            System.out.println("GroupedObservable onNext: " + integer);
                        }

                        @Override
                        public void onError(@NonNull Throwable e) {

                        }

                        @Override
                        public void onComplete() {

                        }
                    });
                }
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {
                System.out.println("groupBy onComplete: ");
            }
        });


        mapObserver().subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("mapObserver subscribe");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                System.out.println("mapObserver onNext：" + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                System.out.println("mapObserver onError：" + e.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("mapObserver onComplete：");
            }
        });


        castObserver().subscribe(new Observer<Dog>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("castObserver subscribe");
            }

            @Override
            public void onNext(@NonNull Dog dog) {
                System.out.println("castObserver onNext " + dog.getName());
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {
                System.out.println("castObserver onComplete ");
            }
        });


        scabObserver().subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("scabObserver subscribe");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                System.out.println("scabObserver onNext " + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                System.out.println("scabObserver onError");
            }

            @Override
            public void onComplete() {
                System.out.println("scabObserver onComplete");
            }
        });

        windowObserver().subscribe(new Observer<Observable<Integer>>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("windowObserver subscribe");
            }

            @Override
            public void onNext(@NonNull Observable<Integer> integerObservable) {
                System.out.println("windowObserver onNext " + integerObservable.getClass().getSimpleName());
                integerObservable.subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {

                    }

                    @Override
                    public void onNext(@NonNull Integer integer) {
                        System.out.println("window integerObservable onNext " + integer);
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    private static Observable<List<Integer>> bufferObserver(){
        return Observable.just(1,2,3,4,5,6,7,8,9,10).buffer(2,3);
    }

    private static Observable<String> flatMapObserver(){
        return Observable.just(1,2,3).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Throwable {
                return Observable.just("flatMap" + integer);
            }
        });
    }


    private static Observable<String> flatMapIterableObserver(){
        return Observable.just(1,2,3).flatMapIterable(new Function<Integer, Iterable<? extends String>>() {
            @Override
            public Iterable<? extends String> apply(Integer integer) throws Throwable {
                ArrayList<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("flatMapIterable" + integer);
                }
                return list;
            }
        });
    }


    private static Observable<GroupedObservable<Integer, Integer>> groupByObserver() {
        return Observable.just(1,2,3,4,5,6,7,8,9).groupBy(new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer integer) throws Throwable {
                return integer % 2;
            }
        });
    }


    private static Observable<Integer> mapObserver() {
        return Observable.just(1,2,3).map(new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer integer) throws Throwable {
                return integer * 10;
            }
        });
    }


    private static Observable<Dog> castObserver() {
        Animal a = new Dog();
        return Observable.just(a).cast(Dog.class);
    }

    private static Observable<Integer> scabObserver() {
        ArrayList<Integer> list = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            list.add(2);
        }
        return Observable.fromIterable(list).scan(new BiFunction<Integer, Integer, Integer>() {
            @Override
            public Integer apply(Integer integer, Integer integer2) throws Throwable {
                return integer * integer2;
            }
        });
    }

    private static Observable<Observable<Integer>> windowObserver() {
        return Observable.just(1,2,3,4,5,6,7,8,9).window(3);
    }
}
